package com.huan.hotitems;


import com.huan.bean.ItemViewCount;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.guava18.com.google.common.collect.Lists;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Comparator;


//实现自定义KeyedProcessFunction
public class TopHotltems extends KeyedProcessFunction<Tuple, ItemViewCount,String> {

    //定义属性，TopN的大小
    private Integer topSize;

    public TopHotltems(Integer topSize){
        this.topSize = topSize;
    }

    //定义列表状态，保存当前当前窗口内所有输出的ItemViewCount
    ListState<ItemViewCount> itemViewCountListState;


    @Override
    public void open(Configuration parameters) throws Exception {
        //将 itemViewCountListState 写入到状态
        itemViewCountListState = getRuntimeContext().getListState( new ListStateDescriptor<ItemViewCount>("item-view-count-list",ItemViewCount.class));

    }


    @Override
    public void processElement(ItemViewCount value, Context ctx, Collector<String> out) throws Exception {
        //每来一条数据,存入List中 并注册定时器
        itemViewCountListState.add( value );
        ctx.timerService().registerProcessingTimeTimer( value.getWindowEnd() + 1 );
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
        //定时器触发，当前已收到的所有数据，排序输出
        ArrayList<ItemViewCount> itemViewCounts = Lists.newArrayList( itemViewCountListState.get().iterator() );
        //开始排序
        itemViewCounts.sort( new Comparator<ItemViewCount>() {
            @Override
            public int compare(ItemViewCount o1, ItemViewCount o2) {
                return o2.getCount().intValue() - o1.getCount().intValue();
            }
        } );

        //将排名信息格式化成String,方便打印输出
        StringBuilder stringBuilder = new StringBuilder();
        stringBuilder.append( "====================================" );
        stringBuilder.append( "窗口结束时间" ).append( new Timestamp( timestamp -1  ) ).append( "\n" );

        //遍历列表，取Top N 输出
        //Math.min (返回两个值最小的一个)
        for (int i = 0;i < Math.min( topSize,itemViewCounts.size() ); i++ ){
            ItemViewCount itemViewCount = itemViewCounts.get( i );
            stringBuilder.append( "NO" ).append( i+1 ).append( ":" ).append( "商品ID = " ).append( itemViewCount.getItemId() )
                    .append( "\t\t" ).append( "商品热门度 = " ).append( itemViewCount.getCount() ).append( "\n" );

        }

        stringBuilder.append( "=====================================\n\n" );

        //控制频率
        Thread.sleep( 1000L );

        //输出
        out.collect( stringBuilder.toString() );

    }
}
